package service

import (
	"encoding/json"
	"github.com/mohae/deepcopy"
	"net/url"
	"strconv"
	"strings"
	"sync"
	"time"
	"vrcm/src/vrcm/app/admin/models"
	"vrcm/src/vrcm/app/admin/models/live_schedule"
	liveDto "vrcm/src/vrcm/app/admin/service/dto/live"
	dto "vrcm/src/vrcm/app/admin/service/dto/live_schedule"
	StreamHubDto "vrcm/src/vrcm/app/admin/service/dto/stream_hub"
	liveService "vrcm/src/vrcm/app/admin/service/live"
	streamHubService "vrcm/src/vrcm/app/admin/service/stream_hub"
	"vrcm/src/vrcm/common/service"
	ext "vrcm/src/vrcm/envconfig"
)

type TransCodeTaskService struct {
	service.Service
}

// CreateChannelTransCodeTask 创建转码任务
func (e *TransCodeTaskService) CreateChannelTransCodeTask(wg *sync.WaitGroup, workTotalNum int, partNum *int, tasks []live_schedule.ChannelTransCodeTask, upMap map[int]live_schedule.TrafficOn, offMap map[int]live_schedule.TrafficOff, basic live_schedule.ChannelBasic, mediaProcess live_schedule.MediaProcess, sourcesMap map[int]live_schedule.SourceBasic) error {

	channelTaskService := ChannelTaskService{}
	channelTaskService.Log = e.Log
	channelTaskService.Orm = e.Orm
	var m sync.Mutex
	liveService := &liveService.LiveBasic{}
	streamHubService := &streamHubService.StreamHubService{}
	streamHubService.Log = e.Log
	liveService.Log = e.Log
	liveBasicService := SourceBasicService{}
	liveBasicService.Log = e.Log
	liveBasicService.Orm = e.Orm
	staticsService := &ChannelStatisticsService{}
	staticsService.Orm = e.Orm
	staticsService.Log = e.Log
	ids := make([]int, 0, 5)
	for _, v := range tasks {
		ids = append(ids, v.SourceId)
	}
	now := time.Now().Unix()

	// 创建任务通道
	//taskCh := make(chan live_schedule.ChannelTransCodeTask, 3)
	// 创建等待组
	//var wg sync.WaitGroup
	//wg.Add(len(tasks))
	// 启动协程执行任务
	//go func() {
	//	for i := 0; i < len(tasks); i++ {
	//		taskCh <- tasks[i] // 将任务放入通道
	//	}
	//	close(taskCh) // 关闭通道，表示任务已经全部放入通道
	//}()
	//if len(taskCh) > 0 {
	//	err := e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", basic.Id).Update("channel_state", 4).Error
	//	if err != nil {
	//		return err
	//	}
	//}

	e.Log.Trace("信源信息为：", sourcesMap)
	// 启动多个协程处理任务
	for i := 0; i < len(tasks); i++ {
		wg.Add(1)
		go func(v live_schedule.ChannelTransCodeTask, wg *sync.WaitGroup) {
			defer wg.Done()

			v1 := live_schedule.ChannelTransCodeTask{}
			//获取上车点
			up := upMap[v.TrafficOnId]

			//获取下车点配置
			off := offMap[v.TrafficOffId]

			//如果是srt拉流或者httpts走的还是内网
			if off.TrafficOffType == 3 || (off.TrafficOffType == 4 && off.StreamType == 2) {
				off.Area = 2
			}

			//信源
			source, ok := sourcesMap[v.SourceId]
			if !ok {
				m.Unlock()
				*partNum++
				m.Unlock()
				e.Log.Errorf("信源不存在")
				return
			}

			v.ExtraInfo = mediaProcess.ExtraInfo
			err := e.CreateSourceTransCodeTask(source, &v)
			if err != nil {
				e.Log.Errorf("CreateSourceTransCodeTask error:", err)
				return
			}
			e.Log.Trace("当前使用信源：", source.SourceName)
			//pullUrlPart := ""
			pullDomain := ""
			if (len(v.PushUrl) <= 0 && off.StreamType == 1) || (off.IsCustom == 1 && len(v.CustomOutUrl) <= 0) {
				e.Log.Trace("不存在转推地址或者拉流地址")
			} else {
				req, autoReq, err := e.GenerateScheduleParam(up, off, mediaProcess, &v, source)
				if err != nil {
					e.Log.Errorf("请求参数失败")
					return
				}
				hub, err := streamHubService.CreateStreamHub(&req)
				if err != nil {
					e.Log.Errorf("CreateStreamHub error:%s \r\n", err)
					//tx.Rollback()
					//return err
				}

				if hub != nil {
					if len(hub.Input.Sources) > 0 {
						v.InUrl = hub.Input.Sources[0].UdpAddress
						v.InNodeId = hub.Input.Sources[0].Id
					}
					//如果是外网拉流就不需要再次接收返回的输出url
					if len(hub.Output.Targets) > 0 {
						if !((off.Area == 1 || off.Area == 3) && off.StreamType == 2) {
							v.OutUrl = hub.Output.Targets[0].Url
						}
						//如果下车点是httpts取AdditionalPublishUrl
						if off.TrafficOffType == 3 {
							v.OutUrl = hub.Output.Targets[0].AdditionalPublishUrl
						}
						v.OutNodeId = hub.Output.Targets[0].Id
						if hub.Output.Targets[0].RedefinePlayUrlEnable == "1" {
							path := hub.Output.Targets[0].RedefinePlayUrlParams.Path + "/" + hub.Output.Targets[0].RedefinePlayUrlParams.Sub + ".m3u8"
							v.CustomOutUrl = path
						}
					}
					v.StateUpdateTime = now
					v.TaskState = 3
					v.TaskId = hub.TaskId
					v.OutputStreamId = hub.OutputStreamName
					v.UpdateTime = now
				}

				//自动调度内网情况
				if off.Area == 3 && off.StreamType == 2 && (off.TrafficOffType == 1 || off.TrafficOffType == 2 || off.TrafficOffType == 6) {
					hub1, err := streamHubService.CreateStreamHub(&autoReq)
					if err != nil {
						e.Log.Errorf("CreateStreamHub error:%s \r\n", err)
						//tx.Rollback()
						//return err
					}
					if hub1 != nil {
						v1 = deepcopy.Copy(v).(live_schedule.ChannelTransCodeTask)
						if len(hub1.Input.Sources) > 0 {
							v1.InUrl = hub1.Input.Sources[0].UdpAddress
							v1.InNodeId = hub1.Input.Sources[0].Id
						}

						//如果是外网拉流就不需要再次接收返回的输出url
						if len(hub1.Output.Targets) > 0 {
							v1.OutUrl = hub1.Output.Targets[0].Url
							pullDomain = v1.OutUrl[0:strings.LastIndex(v1.OutUrl, "/")]
							v1.OutNodeId = hub1.Output.Targets[0].Id
							if hub1.Output.Targets[0].RedefinePlayUrlEnable == "1" {
								path := hub1.Output.Targets[0].RedefinePlayUrlParams.Path + "/" + hub1.Output.Targets[0].RedefinePlayUrlParams.Sub + ".m3u8"
								v1.CustomOutUrl = path
							}
						}
						//如果下车点是httpts取AdditionalPublishUrl
						if off.TrafficOffType == 3 {
							v.OutUrl = hub1.Output.Targets[0].AdditionalPublishUrl
						}
						v.OutUrl = pullDomain + "/" + v.OutUrl[strings.LastIndex(v.OutUrl, "/")+1:]
						v1.StateUpdateTime = now
						v1.TaskState = 3
						v1.TaskId = hub1.TaskId
						v1.Area = 2
						v1.OutputStreamId = hub1.OutputStreamName
						v1.UpdateTime = now
					}
				}

			}

			//如果通道状态是启用 就开启协助任务
			e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", basic.Id).Find(&basic)
			if basic.ChannelState == 1 {
				if source.Kind == 2 && len(source.TaskId) > 0 {
					err = liveBasicService.StartSource(&dto.StartSourceReq{TenantId: basic.TenantId, Id: source.Id})
					if err != nil {
						e.Log.Errorf("StartStreamHub error:%s \r\n", err)
						//tx.Rollback()
						//return err
					}
				}
				err = streamHubService.StartStreamHub(&StreamHubDto.StreamHubReq{TenantId: basic.TenantId, Id: v.TaskId})
				if err != nil {
					e.Log.Errorf("StartStreamHub error:%s \r\n", err)
					//tx.Rollback()
					//return err
				}
				//创建调度统计记录
				err = staticsService.CreateScheduleRecord(basic.TenantId, basic.Id, v.TaskId, 1, now)
				if err != nil {
					e.Log.Errorf("CreateAuditTask error:%s \r\n", err)
				}
				if len(v1.TaskId) > 0 {
					err = streamHubService.StartStreamHub(&StreamHubDto.StreamHubReq{TenantId: basic.TenantId, Id: v1.TaskId})
					if err != nil {
						e.Log.Errorf("StartStreamHub error:%s \r\n", err)
						//tx.Rollback()
						//return err
					}
					//创建调度统计记录
					err = staticsService.CreateScheduleRecord(basic.TenantId, basic.Id, v1.TaskId, 1, now)
					if err != nil {
						e.Log.Errorf("CreateAuditTask error:%s \r\n", err)
					}
				}

			}

			//设置黑白名单
			taskIds := make([]string, 0, 5)
			taskIds = append(taskIds, v.OutNodeId)
			taskIds = append(taskIds, v1.OutNodeId)
			channelTaskService.ProcessBlackWhite(taskIds, off)

			//存储任务历史记录
			if off.StreamType == 2 && len(v.TaskId) > 0 && len(v.OutUrl) > 0 {
				history := live_schedule.ScheduleTaskHistory{}
				history.TenantId = v.TenantId
				history.ChannelId = v.ChannelId
				history.PullUrl = v.OutUrl
				history.OffId = v.TrafficOffId
				history.CreateTime = now
				history.UpdateTime = now
				err = e.Orm.Save(&history).Error
				if err != nil {
					e.Log.Errorf("Save ScheduleTaskHistory error:%s \r\n", err)
				}
			}
			//m.Lock()
			//*partNum++
			err = e.Orm.Save(&v).Error
			if err != nil {
				e.Log.Errorf("Save channel task error:%s \r\n", err)
			}
			if len(v1.TaskId) > 0 {
				err = e.Orm.Save(&v1).Error
				if err != nil {
					e.Log.Errorf("Save channel task error:%s \r\n", err)
				}
			}

			//err = e.CalculateProgress(*partNum, workTotalNum, basic)
			//if err != nil {
			//	e.Log.Trace("CalculateProgress error：", err)
			//}
			//e.Log.Trace("当前运行的任务个数：", *partNum, workTotalNum)
			//m.Unlock()

		}(tasks[i], wg)
	}
	return nil
}

func (e *TransCodeTaskService) CreateSourceTransCodeTask(sourceBasic live_schedule.SourceBasic, task *live_schedule.ChannelTransCodeTask) error {
	//now := time.Now().Unix()
	streamHubService := &streamHubService.StreamHubService{}
	streamHubService.Log = e.Log

	var c dto.CreateTranscodeReq
	var err error
	schedules := make([]live_schedule.ScheduleRecord, 0, 5)
	//创建转码任务
	input := StreamHubDto.TransferCodeInput{}
	output := StreamHubDto.TransferCodeOutput{}
	input.Sources = make([]StreamHubDto.TransCodeSource, 0, 5)
	source := StreamHubDto.TransCodeSource{
		Url: sourceBasic.OutUdpUrl,
		//UdpAddress: c.CustomTransCodeUrl,
	}

	err = json.Unmarshal([]byte(task.ExtraInfo), &c)
	if err != nil {
		e.Log.Errorf("Unmarshal ExtraInfo error:%s \r\n", err)
		return err
	}
	if c.IsCustom == 1 {
		source.UdpAddress = c.CustomTransCodeUrl
	}
	input.Sources = append(input.Sources, source)
	input.Format = "udp"
	input.Type = "push"
	input.Policy = StreamHubDto.Policy{}
	input.Policy.Mode = "manual"
	input.Active = 0

	output.Template = c.Template
	req := StreamHubDto.TransCodeReq{
		TenantId:        c.TenantId,
		Name:            "",
		Input:           input,
		Output:          output,
		Type:            "live",
		CallbackAddress: "",
		NgvsaServiceUrl: ext.ExtConfig.ServiceConfig.NgvsaServiceUrl,
		VrcServiceUrl:   ext.ExtConfig.ServiceConfig.VrcServiceUrl,
	}
	req.CallbackAddress = ext.ExtConfig.ServiceConfig.VrcmServiceUrl + "/api/v1/source-basic/transcode-call-back"
	res, err := streamHubService.CreateTransCode(&req)
	if err != nil {
		e.Log.Errorf("CreateTransCode error:%s \r\n", err)
		return err
	}
	if res != nil {
		//存储转码任务
		for _, s := range res.Input.Sources {
			task.TransTaskId = res.Id
			task.InUrl = s.UdpAddress
			task.OutUrl = s.UdpAddress
		}
	}
	if len(schedules) > 0 {
		err = e.Orm.Model(live_schedule.ScheduleRecord{}).Save(&schedules).Error
		if err != nil {
			e.Log.Errorf("save ScheduleRecord error:%s \r\n", err)
			return err
		}
	}

	return nil
}

func (e *TransCodeTaskService) GenerateScheduleParam(up live_schedule.TrafficOn, off live_schedule.TrafficOff, mediaProcess live_schedule.MediaProcess, task *live_schedule.ChannelTransCodeTask, source live_schedule.SourceBasic) (StreamHubDto.StreamHubReq, StreamHubDto.StreamHubReq, error) {

	pullUrlPart := ""
	liveService := &liveService.LiveBasic{}
	liveService.Log = e.Log
	target := StreamHubDto.Target{}
	target.Format = protocols[off.TrafficOffType]
	target.Id = task.OutNodeId
	//输入参数
	sourceParam := StreamHubDto.Source{}

	////使用转码源还是原始源
	//if up.StreamType == 1 {
	//	e.Log.Trace("当前使用原始源：", up.SourceId)
	//	sourceParam.Id = task.InNodeId
	//	sourceParam.Url = source.OutUdpUrl
	//} else {
	//	sourceParam.Id = task.InNodeId
	//	sourceParam.Url = up.TransUdpUrl
	//}

	sourceParam.Id = task.InNodeId
	sourceParam.Url = task.InUrl
	req := StreamHubDto.StreamHubReq{}

	if len(task.TaskId) > 0 {
		req.Id = task.TaskId
	}

	req.Name = ""
	req.Type = "live"
	req.TenantId = task.TenantId
	req.Description = ""
	req.Input = StreamHubDto.Input{
		Format: protocols[5],
		Active: 0,
		Policy: StreamHubDto.Policy{
			Mode: "manual",
		},
		Type:    "push",
		Sources: []StreamHubDto.Source{sourceParam},
	}

	//第三方拉流+内网 输出类型为pull 其他情况为push
	if off.StreamType == 2 && off.Area == 2 {
		target.Type = "pull"
	} else {
		target.Type = "push"
	}
	delay := StreamHubDto.Delay{
		Enable:   "no",
		Duration: 10,
	}
	if mediaProcess.PlayType == 2 {
		delay.Enable = "yes"
		delay.Duration = mediaProcess.Delay * 1000
	}
	req.CallbackAddress = ext.ExtConfig.ServiceConfig.VrcmServiceUrl + "/api/v1/channel-task/convert-task-callback"

	target1 := deepcopy.Copy(target).(StreamHubDto.Target)
	target1.Type = "pull"
	//外网+第三方拉流需要先调用全球直播
	if (off.Area == 3 || off.Area == 1) && off.StreamType == 2 {
		target.Format = "rtmp"
		streamReq := e.GenerateInsertLiveParam(task.TenantId)
		res, err := liveService.CreateLiveStream(streamReq)
		if err != nil {
			e.Log.Errorf("CreateLiveStream error:%s \r\n", err)
			return req, req, err
		}

		task.LiveId = res.LiveId
		//记录额外信息
		extraMap := make(map[string]string)
		extraMap["rtmp"] = res.RtmpUrl
		extraMap["hls"] = res.HlsUrl
		extraMap["flv"] = res.FlvUrl

		data, err := json.Marshal(extraMap)
		if err != nil {
			e.Log.Errorf("Marshal error:%s \r\n", err)
			return req, req, err
		}
		task.ExtraInfo = string(data)
		pushes := strings.Split(res.PushUrl, models.Delimiter)
		if len(pushes) > 0 {
			target.Url = pushes[0]
			task.PushUrl = pushes[0]
		}

		task.OutUrl = extraMap[protocols[off.TrafficOffType]]

		u, err := url.Parse(task.OutUrl)
		if err != nil {
			return req, req, err
		}
		path := strings.TrimPrefix(u.Path, "/")
		streamName := strings.Split(path, "/")[1]
		queryString := strings.Split(streamName, ".")[0]
		pullUrlPart = queryString
		//如果是外网并且拉流地址
		if off.Area == 3 && off.StreamType == 2 {
			task.Area = 1
		}

	} else if off.StreamType == 1 {
		target.Url = task.PushUrl
	}
	target.RedefinePlayUrlEnable = "2"
	params := StreamHubDto.RedefinePlayUrlParams{}
	target.RedefinePlayUrlParams = params

	//自动调度请求参数
	autoReq := deepcopy.Copy(req).(StreamHubDto.StreamHubReq)
	//如果是自定义拉流地址
	if off.StreamType == 2 && off.TrafficOffType == 2 && (off.Area == 2 || off.Area == 3) && off.IsCustom == 1 {
		arr := strings.Split(task.CustomOutUrl, "/")
		if off.Area == 3 {
			if len(arr) > 1 {
				target1.RedefinePlayUrlEnable = "1"
				params.Path = arr[0]
				index := strings.Index(arr[1], ".")
				params.Sub = arr[1][0:index]
				target1.RedefinePlayUrlParams = params
			}
		} else {
			if len(arr) > 1 {
				target.RedefinePlayUrlEnable = "1"
				params.Path = arr[0]
				index := strings.Index(arr[1], ".")
				params.Sub = arr[1][0:index]
				target.RedefinePlayUrlParams = params
			}
		}

	}
	//创建协助任务
	req.Output = StreamHubDto.Output{
		Targets: []StreamHubDto.Target{target},
		Delay:   delay,
		Acl:     StreamHubDto.Acl{},
	}
	autoReq.Output = StreamHubDto.Output{
		Targets: []StreamHubDto.Target{target1},
		Delay:   delay,
		Acl:     StreamHubDto.Acl{},
	}
	autoReq.OutputStreamName = pullUrlPart
	return req, autoReq, nil
}

func (e *TransCodeTaskService) StartTransCodeTasks(wg *sync.WaitGroup, tasks []live_schedule.ChannelTransCodeTask, workNum int, partNum *int, param map[string]interface{}, sourceMap map[int]live_schedule.SourceBasic) error {
	// 创建任务通道
	//taskCh := make(chan live_schedule.ChannelTransCodeTask, 3)
	streamHubService := &streamHubService.StreamHubService{}
	streamHubService.Log = e.Log
	sourceBasicService := &SourceBasicService{}
	sourceBasicService.Log = e.Log
	sourceBasicService.Orm = e.Orm
	channelStatisticsService := &ChannelStatisticsService{}
	channelStatisticsService.Log = e.Log
	channelStatisticsService.Orm = e.Orm
	//// 启动协程执行任务
	//go func() {
	//	for i := 0; i < len(tasks); i++ {
	//		taskCh <- tasks[i] // 将任务放入通道
	//	}
	//	close(taskCh) // 关闭通道，表示任务已经全部放入通道
	//}()
	//var m sync.Mutex
	now := time.Now().Unix()
	for i := 0; i < len(tasks); i++ {
		wg.Add(1)
		go func(v live_schedule.ChannelTransCodeTask, wg *sync.WaitGroup) {
			defer wg.Done()

			up := sourceMap[v.SourceId]
			if up.Kind == 2 {
				err := sourceBasicService.StartSource(&dto.StartSourceReq{TenantId: v.TenantId, Id: v.SourceId})
				if err != nil {
					e.Log.Errorf("StartSource error:%s \r\n", err)
				}
			}
			if len(v.TransTaskId) > 0 {
				//开启协助任务
				err := streamHubService.StartTransCode(&StreamHubDto.TransCodeReq{TenantId: v.TenantId, Id: v.TransTaskId})
				if err != nil {
					e.Log.Errorf("StartTransCode error:%s \r\n", err)
				}
				//创建调度统计记录
				err = channelStatisticsService.CreateScheduleRecord(v.TenantId, v.ChannelId, v.TransTaskId, 2, now)
				if err != nil {
					e.Log.Errorf("CreateScheduleRecord error:%s \r\n", err)
				}
			}
			if len(v.TaskId) > 0 {
				//开启协助任务
				err := streamHubService.StartStreamHub(&StreamHubDto.StreamHubReq{TenantId: v.TenantId, Id: v.TaskId})
				if err != nil {
					e.Log.Errorf("StartStreamHub error:%s \r\n", err)
					//tx.Rollback()
					//return err
				}
				//创建调度统计记录
				err = channelStatisticsService.CreateScheduleRecord(v.TenantId, v.ChannelId, v.TaskId, 1, now)
				if err != nil {
					e.Log.Errorf("CreateAuditTask error:%s \r\n", err)
				}
			}
			//m.Lock()
			//*partNum++
			//result := float64(*partNum) / float64(workNum)
			//formattedResult := strconv.FormatFloat(result, 'f', 2, 64)
			//param["progress"] = formattedResult
			//if workNum == *partNum {
			//	err := e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", v.ChannelId).Updates(param).Error
			//	if err != nil {
			//		e.Log.Errorf("ChannelBasic Updates error:%s \r\n", err)
			//	}
			//} else {
			//	err := e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", v.ChannelId).Update("progress", formattedResult).Error
			//	if err != nil {
			//		e.Log.Errorf("ChannelBasic Updates error:%s \r\n", err)
			//	}
			//}
			//e.Log.Trace("当前运行的任务个数：", *partNum, workNum)
			//m.Unlock()
		}(tasks[i], wg)
	}

	return nil

}

func (e *TransCodeTaskService) StopTransCodeTasks(wg *sync.WaitGroup, tasks []live_schedule.ChannelTransCodeTask, workNum int, partNum *int, param map[string]interface{}, sourceMap map[int]live_schedule.SourceBasic) error {
	// 创建任务通道
	//taskCh := make(chan live_schedule.ChannelTransCodeTask, 3)
	streamHubService := &streamHubService.StreamHubService{}
	streamHubService.Log = e.Log
	sourceBasicService := &SourceBasicService{}
	sourceBasicService.Log = e.Log
	sourceBasicService.Orm = e.Orm
	//// 启动协程执行任务
	//go func() {
	//	for i := 0; i < len(tasks); i++ {
	//		taskCh <- tasks[i] // 将任务放入通道
	//	}
	//	close(taskCh) // 关闭通道，表示任务已经全部放入通道
	//}()

	//var mu sync.Mutex
	now := time.Now().Unix()
	for i := 0; i < len(tasks); i++ {
		wg.Add(1)
		go func(v live_schedule.ChannelTransCodeTask, wg *sync.WaitGroup) {
			defer wg.Done()
			up := sourceMap[v.SourceId]
			if up.Kind == 2 {
				err := sourceBasicService.StopSource(&dto.StopSourceReq{TenantId: v.TenantId, TaskId: up.TaskId})
				if err != nil {
					e.Log.Errorf("StopSource error:%s \r\n", err)
				}
			}
			//停止协助任务
			if len(v.TaskId) > 0 {
				err := streamHubService.StopStreamHub(&StreamHubDto.StreamHubReq{Id: v.TaskId, TenantId: v.TenantId})
				if err != nil {
					e.Log.Errorf("ProcessWorkflow StopStreamHub error:%s \r\n", err)
				}
			}

			//转码
			if len(v.TransTaskId) > 0 {
				err := streamHubService.StopTransCode(&StreamHubDto.TransCodeReq{Id: v.TransTaskId, TenantId: v.TenantId})
				if err != nil {
					e.Log.Errorf("StopTransCode error:%s \r\n", err)
				}
			}
			//mu.Lock()
			//更新通道任务状态为离线
			param1 := make(map[string]interface{})
			param1["update_time"] = now
			param1["task_state"] = 3
			err := e.Orm.Model(live_schedule.ChannelTask{}).Where("tenant_id=? and channel_id=? and task_id=?", v.TenantId, v.ChannelId, v.TransTaskId).Updates(&param1).Error
			if err != nil {
				e.Log.Errorf("ProcessWorkflow Updates ChannelTask error:%s \r\n", err)
			}
			//更新任务记录结束时间
			param2 := make(map[string]interface{})
			param2["end_time"] = now
			err = e.Orm.Model(live_schedule.ScheduleRecord{}).Where("channel_task_id = ? and end_time = 0 and type =1", v.TaskId).Updates(&param2).Error
			if err != nil {
				e.Log.Errorf("Updates ScheduleRecord error:%s \r\n", err)
			}

			//更新转码任务记录结束时间
			err = e.Orm.Model(live_schedule.ScheduleRecord{}).Where("channel_task_id = ? and end_time = 0 and type =2", v.TransTaskId).Updates(&param2).Error
			if err != nil {
				e.Log.Errorf("Updates ScheduleRecord error:%s \r\n", err)
			}

			//*partNum++
			//result := float64(*partNum) / float64(workNum)
			//formattedResult := strconv.FormatFloat(result, 'f', 2, 64)
			//param["progress"] = formattedResult
			//if *partNum == workNum {
			//	err = e.Orm.Model(live_schedule.ChannelBasic{}).Where("id=?", v.ChannelId).Updates(param).Error
			//	if err != nil {
			//		e.Log.Errorf("Update ChannelTask error:%s \r\n", err)
			//	}
			//} else {
			//	err = e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", v.ChannelId).Update("progress", formattedResult).Error
			//	if err != nil {
			//		e.Log.Errorf("ChannelBasic Updates error:%s \r\n", err)
			//	}
			//}
			//e.Log.Trace("当前运行的任务个数：", *partNum, workNum)
			//mu.Unlock()
		}(tasks[i], wg)
	}

	return nil

}

func (e *TransCodeTaskService) DeleteTransCodeTask(wg *sync.WaitGroup, workTotalNum int, partNum *int, tasks []live_schedule.ChannelTransCodeTask, sourcesMap map[int]live_schedule.SourceBasic, basic live_schedule.ChannelBasic, liveService *liveService.LiveBasic, streamHubService *streamHubService.StreamHubService, liveBasicService SourceBasicService) error {

	//var m sync.Mutex
	liveBasicService.Orm = e.Orm
	// 创建任务通道
	taskCh := make(chan live_schedule.ChannelTransCodeTask, 3)
	go func() {
		for i := 0; i < len(tasks); i++ {
			taskCh <- tasks[i] // 将任务放入通道
		}
		close(taskCh) // 关闭通道，表示任务已经全部放入通道
	}()

	now := time.Now().Unix()
	for i := 0; i < 3; i++ {
		wg.Add(1)
		go func(v live_schedule.ChannelTransCodeTask, wg *sync.WaitGroup) {
			defer wg.Done()
			//source := sourcesMap[v.SourceId]
			////删除自定义
			//if source.Kind == 2 && len(source.TaskId) > 0 {
			//	err := liveBasicService.DeleteCustomSource(source)
			//	if err != nil {
			//		e.Log.Errorf("DeleteCustomSource error:%s \r\n", err)
			//	}
			//}
			//删除协转任务流
			if len(v.TaskId) > 0 {
				err := streamHubService.RemoveStreamHub(&StreamHubDto.StreamHubReq{TenantId: v.TenantId, Id: v.TaskId})
				if err != nil {
					e.Log.Errorf("RemoveStreamHub error:%s \r\n", err)
				}
				param := make(map[string]interface{})
				param["end_time"] = now
				err = e.Orm.Model(live_schedule.ScheduleRecord{}).Where("channel_task_id = ? and end_time = 0", v.TaskId).Updates(&param).Error
				if err != nil {
					e.Log.Errorf("updateScheduleRecord error:%s \r\n", err)
				}
			}
			//删除转码任务流
			if len(v.TaskId) > 0 {
				err := streamHubService.StopTransCode(&StreamHubDto.TransCodeReq{TenantId: v.TenantId, Id: v.TransTaskId})
				if err != nil {
					e.Log.Errorf("StopTransCode error:%s \r\n", err)
				}
				err = streamHubService.DeleteTransCode(&StreamHubDto.TransCodeReq{TenantId: v.TenantId, Id: v.TransTaskId})
				if err != nil {
					e.Log.Errorf("DeleteTransCode error:%s \r\n", err)
				}
				param := make(map[string]interface{})
				param["end_time"] = now
				err = e.Orm.Model(live_schedule.ScheduleRecord{}).Where("channel_task_id = ? and end_time = 0", v.TransTaskId).Updates(&param).Error
				if err != nil {
					e.Log.Errorf("updateScheduleRecord error:%s \r\n", err)
				}
			}
			err := e.Orm.Delete(&v).Error
			if err != nil {
				e.Log.Errorf("SaveWorkflow_Delete error:%s \r\n", err)
			}
			//m.Lock()
			//*partNum++
			//err = e.CalculateProgress(*partNum, workTotalNum, basic)
			//if err != nil {
			//	e.Log.Trace("CalculateProgress error：", err)
			//}
			//m.Unlock()
		}(tasks[i], wg)
	}
	return nil

}

func (e *TransCodeTaskService) GenerateInsertLiveParam(tenantId string) *liveDto.LiveStreamInsertReq {
	streamReq := &liveDto.LiveStreamInsertReq{}
	streamReq.EventName = ""
	streamReq.TenantId = tenantId
	streamReq.ExpireTime = 10 * 365 * 24 * 60 * 60
	streamReq.IMEnable = 0
	streamReq.WebPusherEnable = 0
	streamReq.PullOrPush = ""
	streamReq.AutoRecord = "0"
	streamReq.EventType = "1"
	streamReq.ServiceArea = "cn_mainland"
	return streamReq
}

func (e *TransCodeTaskService) CalculateProgress(partNum int, workTotalNum int, basic live_schedule.ChannelBasic) error {
	result := float64(partNum) / float64(workTotalNum)
	formattedResult := strconv.FormatFloat(result, 'f', 2, 64)
	e.Log.Trace("当前进度为", formattedResult)
	param := make(map[string]interface{})
	param["progress"] = formattedResult
	if workTotalNum <= partNum {
		param["channel_state"] = basic.ChannelState
	}
	err := e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", basic.Id).Updates(&param).Error
	if err != nil {
		e.Log.Trace("update ChannelBasic error：", err)
	}
	return err
}

func (e *TransCodeTaskService) CreateTransCodeTask(node live_schedule.Node, task *live_schedule.ChannelTransTask, inputModels []live_schedule.InputModel, basic live_schedule.ChannelBasic) (live_schedule.ChannelTransTask, error) {
	now := time.Now().Unix()
	streamHubService := &streamHubService.StreamHubService{}
	streamHubService.Log = e.Log
	channelStatisticsService := ChannelStatisticsService{}
	channelStatisticsService.Log = e.Log
	channelStatisticsService.Orm = e.Orm

	//var c dto.CreateTranscodeReq
	var err error
	schedules := make([]live_schedule.ScheduleRecord, 0, 5)
	//创建转码任务
	input := StreamHubDto.TransferCodeInput{}
	output := StreamHubDto.TransferCodeOutput{}
	input.Sources = make([]StreamHubDto.TransCodeSource, 0, 5)
	source := StreamHubDto.TransCodeSource{
		Url: inputModels[0].InputUrl,
		//UdpAddress: c.CustomTransCodeUrl,
	}
	task.InUrl = inputModels[0].InputUrl
	template := StreamHubDto.Template{}
	template.TemplateId = node.Data.Template.TemplateId
	template.Acodec = "aac"
	audioBitrate, _ := strconv.Atoi(node.Data.Template.AacBitrate)
	template.AudioBitrate = audioBitrate
	template.Vcodec = node.Data.Template.VideoCodec
	template.VideoBitrate = node.Data.Template.Bitrate
	template.Width = node.Data.Template.VideoWidth
	template.Height = node.Data.Template.VideoHeight
	template.Fps = node.Data.Template.Fps
	template.Profile = node.Data.Template.Profile
	template.SampleRate = node.Data.Template.AacSamplerate
	output.Template = template
	//err = json.Unmarshal([]byte(task.ExtraInfo), &c)
	//if err != nil {
	//	e.Log.Errorf("Unmarshal ExtraInfo error:%s \r\n", err)
	//	return err
	//}
	if node.Data.IsCustom == 1 {
		source.UdpAddress = node.Data.CustomUrl
	}
	input.Sources = append(input.Sources, source)
	input.Format = "udp"
	input.Type = "push"
	input.Policy = StreamHubDto.Policy{}
	input.Policy.Mode = "manual"
	input.Active = 0

	//output.Template = c.Template
	req := StreamHubDto.TransCodeReq{
		TenantId:        task.TenantId,
		Name:            "",
		Input:           input,
		Output:          output,
		Type:            "live",
		CallbackAddress: "",
		NgvsaServiceUrl: ext.ExtConfig.ServiceConfig.NgvsaServiceUrl,
		VrcServiceUrl:   ext.ExtConfig.ServiceConfig.VrcServiceUrl,
	}
	req.CallbackAddress = ext.ExtConfig.ServiceConfig.VrcmServiceUrl + "/api/v1/source-basic/transcode-call-back"
	if len(task.TaskId) <= 0 {
		//req.Id = task.TaskId
		res, err := streamHubService.CreateTransCode(&req)
		if err != nil {
			e.Log.Errorf("CreateTransCode error:%s \r\n", err)
			return *task, err
		}
		if res != nil {
			//存储转码任务
			for _, s := range res.Input.Sources {
				task.TaskId = res.Id
				task.OutUrl = s.UdpAddress
			}
		}
	} else {
		err = streamHubService.RemoveTransCodeTask(&StreamHubDto.TransCodeReq{TenantId: task.TenantId, Id: task.TaskId})
		if err != nil {
			e.Log.Errorf("RemoveTransCodeTask error:%s \r\n", err)
			return *task, err
		}
		res, err := streamHubService.CreateTransCode(&req)
		if err != nil {
			e.Log.Errorf("CreateTransCode error:%s \r\n", err)
			return *task, err
		}
		if res != nil {
			//存储转码任务
			for _, s := range res.Input.Sources {
				task.TaskId = res.Id
				task.OutUrl = s.UdpAddress
			}
		}
	}

	if len(schedules) > 0 {
		err = e.Orm.Model(live_schedule.ScheduleRecord{}).Save(&schedules).Error
		if err != nil {
			e.Log.Errorf("save ScheduleRecord error:%s \r\n", err)
			return *task, err
		}
	}
	err = e.Orm.Save(&task).Error
	if err != nil {
		e.Log.Trace("save TransTask error：", err)
		return *task, err
	}

	if basic.ChannelState != 2 {
		err = streamHubService.StartTransCode(&StreamHubDto.TransCodeReq{TenantId: basic.TenantId, Id: task.TaskId})
		if err != nil {
			e.Log.Errorf("StartTransCode error:%s \r\n", err)
			return *task, err
		}
		//创建调度统计记录
		err = channelStatisticsService.CreateScheduleRecord(task.TenantId, task.ChannelId, task.TaskId, 2, now)
		if err != nil {
			e.Log.Errorf("CreateScheduleRecord error:%s \r\n", err)
		}
	}

	return *task, nil
}
